package DataStreamApi.Transformation算子;

import domain.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

public class Flink04_keyby {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        List<WaterSensor> waterSensorList = Arrays.asList(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s2", 3L, 3),
                new WaterSensor("s3", 1L, 4),
                new WaterSensor("s1", 11L, 5),
                new WaterSensor("s1", 2L, 6));
        executionEnvironment.setParallelism(3);

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = executionEnvironment.fromCollection(waterSensorList);


        /**
         * 按照ID分组
         * KeyBY不是转换算子，只是对数据进行了重分区，不能设置并行度。
         *
         * 1、KeyBy是对数据进行分组，保证相同key的数据在同一个分区
         * 2、分区是指将数据划分到不同的区域，一个子任务可以认为是一个分区。
         *
         */
        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = waterSensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        });


        waterSensorStringKeyedStream.print();

        executionEnvironment.execute();

    }
}